package com.zhao.biz.tag.save2redis

import ch.hsr.geohash.GeoHash
import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.zhao.utils.{BaiduLBSHandler, HttpUtils, JedisClusterUtils}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Dataset, Row, SparkSession}
import redis.clients.jedis.JedisCluster

/**
 * Description: 将广告日志信息中经纬度信息对应的商圈信息解析出来,存储到redis分布式集群中<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * A wet person does not fear the rain. <br/>
 * Date： 2021/1/15 16:49
 *
 * @author 柒柒
 * @version : 1.0
 */

object LongLacBusiness2Redis {

  def main(args: Array[String]): Unit = {
//    if (args.length != 1) {
//      println("argument is wrong!!!")
//      sys.exit(1)
//    }
    val inputPathTmp = "a_data/outputpath"
    //val Array(inputPath) = args
    val Array(inputPath) = Array(inputPathTmp)

    val conf: SparkConf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
      .set("spark.debug.maxToStringFields", "87")

    val spark: SparkSession = SparkSession.builder()
      .config(conf)
      .getOrCreate()

    //获取原始数据中的经纬度信息
    val longAndLat: Dataset[Row] = spark.read.parquet(inputPath)
      .select("_long", "lat")
      .filter("cast(_long as double) >= 73.66 and cast(_long as double) <= 135.05 " +
        "and cast(lat as double) >=3.86 and cast(lat as double) <= 53.55")
      .distinct()

    longAndLat.show(10000)

    //将商圈中信息持久化到redis,便于以后查找
    longAndLat.foreachPartition(itr =>{
      if (!itr.isEmpty) {
        val jedisCluster: JedisCluster = JedisClusterUtils.getJedisCluster
        itr.foreach(t =>{
          val long = t.getAs[String]("_long")
          val lat = t.getAs[String]("lat")
          save2Redis(jedisCluster,long,lat)
        })
        //释放资源
        JedisClusterUtils.releaseResource(jedisCluster)
      }
    })

    //释放资源
    spark.stop()
  }

  /**
   * 保存到redis中
   * @param jedisCluster
   * @param long
   * @param lat
   */
  def save2Redis(jedisCluster: JedisCluster, long: String, lat: String): Unit = {
    //通过百度的逆地址解析,获取到商圈信息
    GeoHash.geoHashStringWithCharacterPrecision(lat.toDouble, long.toDouble,8)
    //进行sn验证
    val business: String = BaiduLBSHandler.parseBusinessTagBy(long, lat)
    if (StringUtils.isNoneBlank(business)){
      println(business)
    }
  }

  def getAddressInfoFromMap(_long: String,lac: String) = {
    val location: String = _long.concat(",").concat(lac)
    val url = "http://restapi.amap.com/v3/geocode/regeo" + "?output=json" + "&key=e90bf061b3e6eeaf6c8355a43fc1ac1e" +
      "&radius=150" + "&extensions=all" + "&location=" + location

    val addressJson: String = HttpUtils.get(url)
    val jsonObj: JSONObject = JSON.parseObject(addressJson)
    val regeocode: JSONObject = jsonObj.getJSONObject("regeocode")

    var realAddress = ""
    try {
      realAddress = regeocode.getString("formatted_address")
    } catch {
      case msg: Exception => {
        try {
          val addressComponent: JSONObject = regeocode.getJSONObject("addressComponent")
          val province: String = addressComponent.getString("province")
          val district: String = addressComponent.getString("district")

          val name: String = regeocode.getJSONObject("businessAreas").getString("name")

          realAddress = province.concat(district).concat(name)
        } catch {
          case msg: Exception => {
            try {
              val businessAreas: JSONObject = regeocode.getJSONArray("businessAreas").getJSONObject(0)
              realAddress = businessAreas.getString("name")
            } catch {
              case msg: Exception => {
                realAddress = " "
              }
            }
          }
        }
      }
    }
    realAddress
  }
}

















